package com.nepu.spark.realtime.util

import java.util

import com.nepu.spark.realtime.constant.KafkaConstant
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import scala.collection.mutable

/**
  * 本类是作为kafka的工具类
  *
  * @author chenshuaijun
  * @create 2022-10-29 14:55
  */
object MyKafkaUtils {
  private val consumerConfigs = mutable.Map(
    // 配置连接的地址
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> MyConfPropertiesUtils(KafkaConstant.BOOTSTRAP_SERVER),
    // 配置key的反序列化的方式
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    // 配置value的反序列化方式
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    // 配置是否自动提交offset
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> "true",
    // 配置offset自动提价的时间间隔，这个默认值就是5s
    //    ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG -> "5000"
    /*
     配置offset的重置策略：这里补充一下，什么时候需要重置offset
     （1）当offset没有初始化的时候需要进行offset的重置，一般是集群刚搭建好
     （2）当集群中offset缺失的时候会进行offset的重置，这种情况的发生是因为，kafka将数据默认保存的时间是7天，等七天之后offset就被删除了
     offset的重置策略有以下四种：
      earliest：是从头开始消费
      latest：之前的数据都不再消费，从现在开始，消费之后生产的数据
      none：会报错
      anything else：会报错
      */
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "latest"

  )

  /**
    * 使用kafka默认的offset消费数据
    *
    * @param ssc     StreamingContext对象
    * @param topic   消费的主题
    * @param groupId 消费者组
    */
  def getKafkaDStream(ssc: StreamingContext, topic: String, groupId: String): InputDStream[ConsumerRecord[String, String]] = {
    // 配置消费者组
    consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
    val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array[String](topic), consumerConfigs))
    dStream
  }

  /**
    * 使用消费指定offset的数据
    *
    * @param ssc     StreamingContext对象
    * @param topic   消费的主题
    * @param groupId 消费者组
    * @param offset  指定消费的offset
    */
  def getKafkaDStream(ssc: StreamingContext, topic: String, groupId: String, offset: Map[TopicPartition, Long]): InputDStream[ConsumerRecord[String, String]] = {
    // 配置消费者组
    consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId)
    val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream(
      ssc,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](Array[String](topic), consumerConfigs, offset))
    dStream
  }


  /**
    * 创建生产者对象
    */
  private var producer: KafkaProducer[String, String] = getProducer()

  def getProducer(): KafkaProducer[String, String] = {
    val map: util.HashMap[String, AnyRef] = new util.HashMap()
    // 配置连接地址
    map.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, MyConfPropertiesUtils(KafkaConstant.BOOTSTRAP_SERVER))
    // 配置key的序列化器
    map.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    // 配置value的序列化器
    map.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    /*
     配置ACK策略:
     ack = 0 表示的是生产者不等待任何确认，直接发数据
     ack = 1 这意味着leader将记录写入本地日志，但不会等待所有follower的完全确认。在这种情况下，如果leader在确认记录后立即失败，但在follower复制它之前，那么记录将丢失
     ack = all 这意味着leader将等待全部同步副本来确认该记录。这保证了只要至少有一个同步副本存活，记录就不会丢失。这是最有力的保证。这相当于acks=-1设置
      */
    map.put(ProducerConfig.ACKS_CONFIG, "all")
    // 开启幂等性
    map.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
    // 还可以配置batchSize-->这个参数默认值是16k，指的是数据在缓冲区中攒够16kb之后，才写入对于的分区
    // linger.ms --> 指的间隔多长时间将数据写入到分区中。他和blockSize是配合使用的

    val producer: KafkaProducer[String, String] = new KafkaProducer[String, String](map)
    producer
  }

  /**
    * 发送数据
    */
  def send(topic: String, value: String): Unit = {
    producer.send(new ProducerRecord[String, String](topic, value))
  }

  /**
    * 关闭生产者对象
    */
  def close(): Unit = {
    producer.close()
  }


  def flush(): Unit = {
    producer.flush()
  }
}
